package com.navinfo.tripanalysis.offline.service;

import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.service.impl.*;
import com.navinfo.tripanalysis.offline.service.impl.huashen.LoadCn6BasicDataInfoServiceImpl;
import com.navinfo.tripanalysis.offline.util.BigDataUtils;
import org.apache.commons.lang3.StringUtils;
import org.reflections.Reflections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.lang.reflect.Constructor;
import java.util.*;

/**
 * Service工厂类
 * @author 沈东生
 */
public class ServiceFactory implements Serializable {
    private static final Logger logger = LoggerFactory.getLogger(ServiceFactory.class);
    /**
     * 全局配置
     */
    private Properties config;
    /**
     * 系统的运行类型
     */
    private RunningTypeEnum runningTypeEnum;
    /**
     * 各种系统运行类型的下接口的不同实现Map<br/>
     * key为：运行环境 + "**" + 接口名
     * value为：实现类的Class
     */
    private static final Map<String, Class> serviceMap = new TreeMap<>();

    static{
        synchronized (serviceMap){
            //扫描工程包下添加了RunningType注解的class，注册系统运行时各接口的类实现
            new Reflections("com.navinfo.tripanalysis.offline.service.impl")
                    .getTypesAnnotatedWith(RunningType.class).
                    forEach(clazz ->{
                        RunningType runningType = clazz.getAnnotation(RunningType.class);
                        serviceMap.put(runningType.value().getType()+ "**"+runningType.interfaceClass().getSimpleName(), clazz);
                    });

            StringBuilder sb = new StringBuilder();
            serviceMap.forEach((key, value) -> sb.append(key+"-->"+value.getName()+"\n"));

            logger.error("系统服务工厂初始化完毕，不同运行环境下的接口实现如下\n{}", sb.toString());
        }
    }

    public ServiceFactory(Properties config, RunningTypeEnum runningTypeEnum) {
        this.config = config;
        this.runningTypeEnum = runningTypeEnum;
    }

    /**
     * Spark算子中可通过该接口创建不同运行环境对应接口的实现类
     * @param runningType
     * @param clazz
     * @param <T>
     * @return
     * @throws IllegalAccessException
     * @throws InstantiationException
     */
    public static <T> T createServiceImpl(RunningTypeEnum runningType, Class<T> clazz) throws IllegalAccessException, InstantiationException {
        if(null == runningType){
            throw new RuntimeException("请传入系统运行类型，1东风，2青汽，3一汽");
        }
        Class aClass;
        synchronized (serviceMap){
            aClass = serviceMap.get(runningType.getType() + "**" + clazz.getSimpleName());
        }

        logger.error("创建{}的实现类{}", clazz, aClass);
        return  (T) aClass.newInstance();
    }

    /**
     * 通过接口获取，当前系统运行类型下的实现类
     */
    private <T> T createServiceImpl(Class<T> clazz) throws IllegalAccessException, InstantiationException {
        if(null == runningTypeEnum){
            throw new RuntimeException("请配置系统运行类型，1东风，2青汽，3一汽");
        }
        Class aClass;
        synchronized (serviceMap){
            aClass = serviceMap.get(runningTypeEnum.getType() + "**" + clazz.getSimpleName());
        }
        logger.error("创建{}的实现类{}", clazz, aClass);
        return (T) aClass.newInstance();
    }

    public Class serviceImplClass(RunningTypeEnum runningType,Class clazz){
        if (null == runningType) {
            throw new RuntimeException("请配置系统运行类型，1东风，2青汽，3一汽");
        }
       return serviceMap.get(runningType.getType() + "**" + clazz.getSimpleName());
    }

    /**
     * 创建加载车辆信息的服务
     */
    public AbstractLoadVehicleInfoService createLoadVehicleInfoService() throws IllegalAccessException, InstantiationException {
        AbstractLoadVehicleInfoService service = (AbstractLoadVehicleInfoService) createServiceImpl(LoadVehicleInfoService.class);
        service.setDriver(config.getProperty("vehicleinfo.jdbc.driver"));
        service.setUrl(config.getProperty("vehicleinfo.jdbc.url"));
        service.setUser(config.getProperty("vehicleinfo.jdbc.username"));
        service.setPass(config.getProperty("vehicleinfo.jdbc.password"));
        return service;
    }
    /**
     * 创建加载国六车辆信息服务
     */
    public LoadCn6BasicDataInfoServiceImpl createLoadCn6BasicDataService() {
        LoadCn6BasicDataInfoServiceImpl service = new LoadCn6BasicDataInfoServiceImpl();
        service.setDbName(config.getProperty("load.cn6.basic.data.hive.dbName"));
        service.setTbName(config.getProperty("load.cn6.basic.data.hive.tbName"));
        service.setSelectHql(config.getProperty("load.cn6.basic.data.hive.selectHql"));
        return service;
    }

    /**
     * 创建加载瓦片信息服务
     */
    public LoadTileService createLoadTileService() {
        LoadTileServiceImpl service = new LoadTileServiceImpl();
        service.setHDFSPath(config.getProperty("load.data.tile.hdfspath"));
        return service;
    }

    /**
     * 创建大数据分批处理服务
     */
    public DataBatchService createDataBatchService() {
        String type = config.getProperty("load.data.batch.type");
        if (StringUtils.equals(type, "bigdata")) {
            DataBatchServiceImpl service = new DataBatchServiceImpl();
            service.setBatchSize(Integer.parseInt(config.getProperty("load.data.batch.bigdata.size")));
            return service;
        }else{
            throw new RuntimeException("不支持DataBatchService类型:" + type);
        }
    }

    /**
     * 创建数据清洗大数据分批处理服务，增加参数为读入不同参数，无实际意义
     */
    public DataBatchService createDataBatchService(boolean flag) {
        if(flag) {
            String type = config.getProperty("load.data.batch.type");
            if (StringUtils.equals(type, "bigdata")) {
                DataBatchServiceImpl service = new DataBatchServiceImpl();
                service.setBatchSize(Integer.parseInt(config.getProperty("load.data.cleaning.batch.bigdata.size")));
                return service;
            } else {
                throw new RuntimeException("不支持DataBatchService类型:" + type);
            }
        }else {
            return null;
        }
    }

    /**
     * 创建加载0f37服务（可hdfs、excel、hive）
     */
    public LoadPoint0F37Service createLoad0F37DataService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("load.data.0f37.type");

        if (StringUtils.equals(type, "hdfs") || StringUtils.equals(type, "hive")) {
            AbstractLoadPoint0F37Service service = (AbstractLoadPoint0F37Service) createServiceImpl(LoadPoint0F37Service.class);
            service.setLoadType(BigDataUtils.LoadType.valueOf(type.toUpperCase()));
            service.setDbName(config.getProperty("load.data.0f37.hive.dbname"));
            service.setTableName(config.getProperty("load.data.0f37.hive.tablename"));
            service.setHdfsPath(config.getProperty("load.data.0f37.hdfs.basepath"));
            service.setSupportProtocolType(Boolean.parseBoolean(config.getProperty("load.data.0f37.supportProtocoltype")));
            return service;
        }else if (StringUtils.equals(type, "excel")) {
            LoadPoint0F37ServiceExcelImpl service = new LoadPoint0F37ServiceExcelImpl();
            service.setPath(config.getProperty("tools.point.excel.basepath"));
            return service;
        }else{
            throw new RuntimeException("不支持Load0F37DataService类型:" + type);
        }
    }

    /**
     * 创建加载0200服务（可hdfs、excel、hive）
     */
    public LoadPoint0200Service createLoad0200DataService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("load.data.0200.type");
        if (StringUtils.equals(type, "hdfs") || StringUtils.equals(type, "hive")) {
            AbstractLoadPoint0200Service service = (AbstractLoadPoint0200Service) createServiceImpl(LoadPoint0200Service.class);
            service.setLoadType(BigDataUtils.LoadType.valueOf(type.toUpperCase()));
            service.setDbName(config.getProperty("load.data.0200.hive.dbname"));
            service.setTableName(config.getProperty("load.data.0200.hive.tablename"));
            service.setHdfsPath(config.getProperty("load.data.0200.hdfs.basepath"));
            service.setSupportSerialNumber(Boolean.parseBoolean(config.getProperty("load.data.0200.supportSerialNumber")));
            return service;
        } else if (StringUtils.equals(type, "excel")) {
            LoadPoint0200ServiceExcelImpl service = new LoadPoint0200ServiceExcelImpl();
            service.setPath(config.getProperty("tools.point.excel.basepath"));
            return service;
        }else{
            throw new RuntimeException("不支持Load0200DataService类型:" + type);
        }
    }

    /**
     * 创建加载0F39服务（可hdfs、hive）
     */
    public LoadPoint0F39Service createLoad0F39DataService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("load.data.0F39.type");
        if (StringUtils.equals(type, "hdfs") || StringUtils.equals(type, "hive")) {
            AbstractLoadPoint0F39Service service = (AbstractLoadPoint0F39Service) createServiceImpl(LoadPoint0F39Service.class);
            service.setLoadType(BigDataUtils.LoadType.valueOf(type.toUpperCase()));
            service.setDbName(config.getProperty("load.data.0F39.hive.dbname"));
            service.setTableName(config.getProperty("load.data.0F39.hive.tablename"));
            service.setHdfsPath(config.getProperty("load.data.0F39.hdfs.basepath"));
            return service;
        } else{
            throw new RuntimeException("不支持Load0F39DataService类型:" + type);
        }
    }

    /**
     * 创建加载0F3D数据服务(支持hive)
     */
    public LoadPoint0F3DService createLoadPoint0F3DService() throws IllegalAccessException, InstantiationException{
        String type = config.getProperty("load.data.0f3d.type");
        if (StringUtils.equals(type, "hdfs") || StringUtils.equals(type, "hive")) {
            AbstractLoadPoint0F3DService service = (AbstractLoadPoint0F3DService) createServiceImpl(LoadPoint0F3DService.class);
            service.setLoadType(BigDataUtils.LoadType.valueOf(type.toUpperCase()));
            service.setDbName(config.getProperty("load.data.0f3d.hive.dbname"));
            service.setTableName(config.getProperty("load.data.0f3d.hive.tablename"));
            service.setHdfsPath(config.getProperty("load.data.0f3d.hdfs.basepath"));
            service.setSupportSerialNumber(Boolean.parseBoolean(config.getProperty("load.data.0f3d.supportSerialNumber")));
            return service;
        }else{
            throw new RuntimeException("不支持Load0F37DataService类型:" + type);
        }
    }

    /**
     * 创建加载通用协议的服务（可hdfs、hive）
     */
    public LoadPointProtocolService createLoadPointService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("load.data.point.type");
        if (StringUtils.equals(type, "hdfs") || StringUtils.equals(type, "hive")) {
            LoadPointProtocolServiceImpl service = new LoadPointProtocolServiceImpl();
            service.setLoadType(BigDataUtils.LoadType.valueOf(type.toUpperCase()));
            service.setDbName(config.getProperty("load.data.point.hive.dbname"));
            service.setTableName(config.getProperty("load.data.point.hive.tablename"));
            service.setHdfsPath(config.getProperty("load.data.point.hdfs.basepath"));
            service.setService(createRowSchemaService());
            return service;
        }else{
            throw new RuntimeException("不支持LoadPointService类型:" + type);
        }
    }

    /**
     * 创建加载通用协议Schema服务
     */
    public PointProtocolConvertService createRowSchemaService() throws InstantiationException, IllegalAccessException {
        PointProtocolConvertService service = createServiceImpl(PointProtocolConvertService.class);
        return service;
    }


    /**
     * 创建位置点重分区、分区内排序服务
     */
    public PointSortGroupService createPointSortGroupService() {
        String type = config.getProperty("pointsSortGroup.type");
        if (StringUtils.equals("type1", type)) {
            PointSortGroupServiceImpl1 service = new PointSortGroupServiceImpl1();
            String removeDuplicate = config.getProperty("pointsSortGroup.removeDuplicate");
            if(StringUtils.isNotEmpty(removeDuplicate)){
                service.setRemoveDuplicate(Boolean.valueOf(removeDuplicate));
            }
            return service;
        } else if (StringUtils.equals("type2", type)) {
            PointSortGroupServiceImpl2 service = new PointSortGroupServiceImpl2();
            String parallel = config.getProperty("pointsSortGroup.type2.parallel");
            if (StringUtils.isNotEmpty(parallel)) {
                service.setParallel(Integer.parseInt(config.getProperty(parallel)));
            }
            return service;
        } else {
            throw new RuntimeException("PointsSortGroupService:" + type);
        }
    }

    /**
     * 数据清洗切分
     * @return
     */
    public PreTripInfoDataCleaningService createPreTripInfoDataCleaningService(RunningTypeEnum runningTypeEnum){
        //注入各车厂行程切分规则
        PreTripInfoDataCleaningServiceImpl service = new PreTripInfoDataCleaningServiceImpl();
        service.setDataCleaningJudgeTripStatusService(createGetJudgeTripStatusService(runningTypeEnum));
        service.setDataCleaningUpdAbnormalPointService(createUpdAbnormalPointService());
        service.setSpeed_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0f37.threshold.speed."+runningTypeEnum.name())));
        service.setRotation_Max_Threshold(Integer.parseInt(
                config.getProperty("load.datacleaning.0f37.threshold.rotation."+runningTypeEnum.name())));
        return service;
    }

    /**
     * 创建行程切分逻辑服务
     * @return
     */
    public DataCleaningJudgeTripStatusService createGetJudgeTripStatusService(RunningTypeEnum runningTypeEnum){
        try{
            DataCleaningJudgeTripStatusService dataCleaningJudgeTripStatusService = createServiceImpl(runningTypeEnum,DataCleaningJudgeTripStatusService.class);
            return dataCleaningJudgeTripStatusService;
        }catch (InstantiationException|IllegalAccessException e) {
            throw new RuntimeException("创建行程切分逻辑服务失败......",e);
        }
    }

    /**
     * 创建用0200数据修正0f37数据服务
     * @return
     */
    public DataCleaningUpdAbnormalPointService createUpdAbnormalPointService(){
        try{
            DataCleaningUpdAbnormalPointService dataCleaningUpdAbnormalPointService
                    = createServiceImpl(DataCleaningUpdAbnormalPointService.class);
            return dataCleaningUpdAbnormalPointService;
        }catch (InstantiationException|IllegalAccessException e) {
            throw new RuntimeException("创建0f37数据修正服务失败......",e);
        }
    }

    /**
     * 创建算法链服务
     * @param algorithmConfig
     * @return
     */
    public AlgorithmChainService createAlgorithmChainService(RunningTypeEnum runningType, Map algorithmConfig) {
        try{
            Class<AlgorithmChainService> aClass = serviceImplClass(runningType, AlgorithmChainService.class);
            Constructor<AlgorithmChainService> constructor = aClass.getConstructor(Map.class);
            return constructor.newInstance(algorithmConfig);
        }catch (Exception e) {
            throw new RuntimeException("创建算法链服务失败......",e);
        }
    }


    /**
     * 创建行程数据落盘服务，可mongo、bigdata
     */
    public List<SaveTripStatisticService> createSaveTripServices() throws IllegalAccessException, InstantiationException {
        return createSaveTripServices("");
    }
    public List<SaveTripStatisticService> createSaveTripServices(String prefix) throws InstantiationException, IllegalAccessException {
        List<SaveTripStatisticService> services = new ArrayList<>();
        List<String> types = CommonUtils.splitStr(config.getProperty(prefix+"save.trip.types"), ",");
        for (String type : types) {
            if (StringUtils.equals("mongo", type)) {
                SaveTripStatisticServiceMongoImpl service = new SaveTripStatisticServiceMongoImpl(config.getProperty(prefix+"save.trip.mongo.collectionNamePrefix"), createTripStatisticConvertService());
                services.add(service);
            } else if (StringUtils.equals("bigdata", type)) {
                SaveTripStatisticServiceHiveImpl service = new SaveTripStatisticServiceHiveImpl(createTripStatisticConvertService());
                setSaveDataBigDataService(service, prefix+"save.trip");
                services.add(service);
            } else {
                throw new RuntimeException("不支持SaveTripService类型:" + type);
            }
        }
        return services;
    }


    /**
     * 创建行程发动机数据落盘服务，目前只bigdata（2021.05.24）
     */
    public List<SaveTripEngineDataService> createSaveTripEngineDataServices() throws IllegalAccessException, InstantiationException {
        return createSaveTripEngineDataServices("");
    }
    public List<SaveTripEngineDataService> createSaveTripEngineDataServices(String prefix) throws IllegalAccessException, InstantiationException {
        List<SaveTripEngineDataService> services = new ArrayList<>();
        List<String> types = CommonUtils.splitStr(config.getProperty(prefix + "save.trip.engine.data.types"), ",");
        for (String type : types) {
            if (StringUtils.equals("mongo", type)) {
//                SaveTripServiceMongoImpl service = new SaveTripServiceMongoImpl();
//                service.setCollectionNamePrefix(config.getProperty(prefix+"save.trip.engine.data.mongo.collectionNamePrefix"));
//                service.setConfig(config);
//                services.add(service);
            } else if (StringUtils.equals("bigdata", type)) {
                SaveTripEngineDataServiceHiveImpl service = new SaveTripEngineDataServiceHiveImpl(createTripEngineConvertService());
                setSaveDataBigDataService(service, "save.trip.engine.data");
                services.add(service);
            } else {
                throw new RuntimeException("不支持SaveTripEngineDataService类型:" + type);
            }
        }
        return services;
    }

    /**
     * 创建加载发动机数据流服务（可hdfs）
     */
    public LoadEngineDataService createLoadEngineDataService() {
        String type = config.getProperty("load.data.engine.type");
        //是否支持serialNumber字段
        boolean supportProtocolType = Boolean.parseBoolean(config.getProperty("load.data.engine.supportProtocoltype"));
        if (StringUtils.equals(type, "hdfs")) {
//            LoadEngineDataServiceHdfsImpl service = new LoadEngineDataServiceHdfsImpl();
//            service.setBasePath(config.getProperty("load.data.engine.hdfs.basepath"));
//            service.setSupportProtocolType(supportProtocolType);
//            return service;
        } else if (StringUtils.equals(type, "hive")) {
            LoadEngineDataServiceHiveImpl service = new LoadEngineDataServiceHiveImpl();
            service.setDbName(config.getProperty("load.data.engine.hive.dbname"));
            service.setTableName(config.getProperty("load.data.engine.hive.tablename"));
            service.setSupportProtocolType(supportProtocolType);
            return service;
        } else if (StringUtils.equals(type, "excel")) {
//            LoadEngineDataServiceHdfsImpl service = new LoadEngineDataServiceHdfsImpl();
//            service.setPath(config.getProperty("tools.point.excel.basepath"));
//            return service;
        }
        throw new RuntimeException("不支持LoadEngineDataService类型:" + type);
    }

    public DataCombineService createDataCombineService() {
        TidPointsDataCombineServiceImpl service = new TidPointsDataCombineServiceImpl();
        return service;
    }

    /**
     * 创建加载国六车辆异常项阈值信息服务
     */
    public LoadCn6ThresholdDataInfoServiceImpl createLoadCn6ThresholdDataService() {
        LoadCn6ThresholdDataInfoServiceImpl service = new LoadCn6ThresholdDataInfoServiceImpl();
        service.setDbName(config.getProperty("load.cn6.threshold.data.hive.dbName"));
        service.setTbName(config.getProperty("load.cn6.threshold.data.hive.tbName"));
        service.setSelectHql(config.getProperty("load.cn6.threshold.data.hive.selectHql"));
        return service;
    }

    private TripStatisticConvertService createTripStatisticConvertService() throws IllegalAccessException, InstantiationException {
        return createServiceImpl(TripStatisticConvertService.class);
    }

    private TripEventConvertService createTripEventConvertService() throws IllegalAccessException, InstantiationException {
        return createServiceImpl(TripEventConvertService.class);
    }

    private TripEngineConvertService createTripEngineConvertService() throws IllegalAccessException, InstantiationException {
        return createServiceImpl(TripEngineConvertService.class);
    }

    /**
     * 创建事件数据落盘服务，可mongo、hbase、bigdata
     */
    public List<SaveTripEventService> createSaveEventServices() throws IllegalAccessException, InstantiationException {
        return createSaveEventServices("");
    }

    public List<SaveTripEventService> createSaveEventServices(String prefix) throws InstantiationException, IllegalAccessException {
        List<SaveTripEventService> services = new ArrayList<>();
        List<String> types = CommonUtils.splitStr(config.getProperty(prefix+"save.event.types"), ",");

        for (String type : types) {
            if (StringUtils.equals("mongo", type)) {
                SaveTripEventServiceMongoImpl service = new SaveTripEventServiceMongoImpl(config.getProperty(prefix+"save.event.mongo.collectionNamePrefix"), createTripEventConvertService());
                services.add(service);
            } else if (StringUtils.equals("hbase", type)) {
                SaveTripEventServiceHBaseImpl service = new SaveTripEventServiceHBaseImpl(createTripEventConvertService());
                service.setTableName(config.getProperty(prefix+"save.event.hbase.tableName"));
                service.setQuorum(config.getProperty("hbase.zookeeper.quorum"));
                service.setPort(config.getProperty("hbase.zookeeper.property.clientPort"));
                service.setParent(config.getProperty("zookeeper.znode.parent"));
                String saveType = config.getProperty(prefix+"save.event.hbase.saveType");
                service.setSaveType(StringUtils.isEmpty(saveType) ? SaveTripEventServiceHBaseImpl.SAVE_TYPE_2 :saveType );
                services.add(service);
            } else if (StringUtils.equals("bigdata", type)) {
                SaveTripEventServiceHiveImpl service = new SaveTripEventServiceHiveImpl(createTripEventConvertService());
                setSaveDataBigDataService(service, prefix+"save.event");
                services.add(service);
            } else {
                throw new RuntimeException("不支持SaveTripService类型:" + type);
            }
        }
        return services;
    }

    /**
     * 创建中间数据落盘服务
     */
    public DrivingDataService createDrivingDataService() {
        String type = config.getProperty("save.drivingdata.type");
        if (StringUtils.equals("bigdata", type)) {
            DrivingDataServiceImpl drivingDataService = new DrivingDataServiceImpl();
            setSaveDataBigDataService(drivingDataService, "save.drivingdata");
            return drivingDataService;
        }else{
            throw new RuntimeException("不支持DrivingDataService类型:" + type);
        }
    }

    /**
     * 创建公共轨迹协议落盘服务
     */
    public SavePointProtocolService createSavePointService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("save.point.bigdata.type");
        if (StringUtils.equals("hive", type)) {
            SavePointProtocolServiceImpl savePointService = new SavePointProtocolServiceImpl();
            savePointService.setService(createRowSchemaService());
            setSaveDataBigDataService(savePointService, "save.point");
            return savePointService;
        }
        return null;
    }

    /**
     * 创建行程统计扩展信息的服务
     */
    public TripStatisticExtendService createTripStatisticExtendService() throws InstantiationException, IllegalAccessException {
        TripStatisticExtendService service = createServiceImpl(TripStatisticExtendService.class);

        if(null!=service && AbstractHiveSaveService.class.isAssignableFrom(service.getClass())){
            AbstractHiveSaveService hiveSaveService = (AbstractHiveSaveService)service;
            String coalesceType = config.getProperty("save.trip.extend.hive.coalesceType");
            if (StringUtils.isNotEmpty(coalesceType)) {
                hiveSaveService.setCoalesceType(BigDataUtils.CoalesceType.valueOf(coalesceType.toLowerCase()));
            }
            String coalesceNumber = config.getProperty("save.trip.extend.hive.coalesceNumber");
            if (StringUtils.isNotEmpty(coalesceNumber)) {
                hiveSaveService.setCoalesceNumber(Integer.valueOf(coalesceNumber));
            }
            hiveSaveService.setSaveType(BigDataUtils.SaveType.HIVE);
            hiveSaveService.setHiveDbName(config.getProperty("save.trip.extend.hive.dbName"));
            hiveSaveService.setHiveTableName(config.getProperty("save.trip.extend.hive.tableName"));
        }

        return service;
    }


    /**
     * 创建公共轨迹协议落盘服务
     */
    public SavePointDataCleaningService createSavePointDataCleaningService() throws IllegalAccessException, InstantiationException {
        String type = config.getProperty("save.point.bigdata.type");
        if (StringUtils.equals("hive", type)) {
            SavePointDataCleaningServiceImpl savePointService = new SavePointDataCleaningServiceImpl();
            savePointService.setService(createRowSchemaService());
            setSaveDataBigDataService(savePointService, "save.point");
            return savePointService;
        }
        return null;
    }

    /**
     * 创建0F39故障数据落盘服务
     */
    public SavePointFaultService createSaveFaultPointService() {
        String type = config.getProperty("save.fault.point.bigdata.type");
        if (StringUtils.equals("hive", type)) {
            SavePointFaultServiceImpl savePointFaultService = new SavePointFaultServiceImpl();
            setSaveDataBigDataService(savePointFaultService, "save.fault.point");
            return savePointFaultService;
        }
        return null;
    }

    private void setSaveDataBigDataService(AbstractHiveSaveService service, String keyPrefix) {
        String coalesceType = config.getProperty(keyPrefix + ".bigdata.coalesceType");
        if (StringUtils.isNotEmpty(coalesceType)) {
            service.setCoalesceType(BigDataUtils.CoalesceType.valueOf(coalesceType.toLowerCase()));
        }
        String coalesceNumber = config.getProperty(keyPrefix + ".bigdata.coalesceNumber");
        if (StringUtils.isNotEmpty(coalesceNumber)) {
            service.setCoalesceNumber(Integer.valueOf(coalesceNumber));
        }
        String insertType = config.getProperty(keyPrefix + ".bigdata.insertType");
        if(StringUtils.isNotEmpty(insertType)){
            service.setInsertType(BigDataUtils.InsertType.valueOf(insertType.toLowerCase()));
        }
        service.setSaveType(BigDataUtils.SaveType.valueOf(config.getProperty(keyPrefix + ".bigdata.type").toUpperCase()));
        service.setHiveDbName(config.getProperty(keyPrefix + ".bigdata.hiveDbName"));
        service.setHiveTableName(config.getProperty(keyPrefix + ".bigdata.hiveTableName"));
        service.setHdfsPath(config.getProperty(keyPrefix + ".bigdata.hdfsBasePath"));
    }

    /**
     * 创建ETL读取行程服务
     */
    public LoadTripStatisticService createLoadTripService() throws InstantiationException, IllegalAccessException {
        LoadTripStatisticServiceImpl service;

        String type = config.getProperty("etl.load.trip.type");
        if (StringUtils.equals("hive", type)){
            service = new LoadTripStatisticServiceImpl(createTripStatisticConvertService());
            service.setLoadType(BigDataUtils.LoadType.HIVE);
            service.setDbName(config.getProperty("etl.load.trip.hive.dbname"));
            service.setTableName(config.getProperty("etl.load.trip.hive.tablename"));
        }else if(StringUtils.equals("hdfs", type)){
            service = new LoadTripStatisticServiceImpl(createTripStatisticConvertService());
            service.setLoadType(BigDataUtils.LoadType.HDFS);
            service.setHdfsPath(config.getProperty("etl.load.trip.hdfs.basepath"));
        }else{
            throw new RuntimeException("不支持加载行程统计信息LoadTripService的类型:" + type);
        }

        return service;
    }

    /**
     * 创建ETL读取事件数据服务
     */
    public LoadTripEventService createLoadEventService() throws InstantiationException, IllegalAccessException {
        LoadTripEventServiceImpl service;

        String type = config.getProperty("etl.load.event.type");
        if (StringUtils.equals("hive", type)){
            service = new LoadTripEventServiceImpl(createTripEventConvertService());
            service.setLoadType(BigDataUtils.LoadType.HIVE);
            service.setDbName(config.getProperty("etl.load.event.hive.dbname"));
            service.setTableName(config.getProperty("etl.load.event.hive.tablename"));
        }else if(StringUtils.equals("hdfs", type)){
            service = new LoadTripEventServiceImpl(createTripEventConvertService());
            service.setLoadType(BigDataUtils.LoadType.HDFS);
            service.setHdfsPath(config.getProperty("etl.load.event.hdfs.basepath"));
        }else{
            throw new RuntimeException("不支持加载行程统计信息LoadTripService的类型:" + type);
        }

        return service;

    }

    /**
     * 创建0200数据清洗服务
     */

    public DataCleaning0200Service createDataCleaning0200Service(){
        DataCleaning0200ServiceImpl service = new DataCleaning0200ServiceImpl();
        service.setSpeed_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0200.threshold.speed."+runningTypeEnum.name())
        ));
        service.setFuel_100km_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0200.threshold.fuel100km."+runningTypeEnum.name())
        ));
        service.setPoint_Time(Integer.parseInt(
                config.getProperty("load.datacleaning.0200.threshold.pointtime."+runningTypeEnum.name())
        ));
//        service.setMileage_Constant(Double.parseDouble(
//                config.getProperty("load.datacleaning.0200.threshold.mileageConstant."+runningTypeEnum.name())
//        ));
        service.setLatLon_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.threshold.LatLonPerSec."+runningTypeEnum.name())
        ));

        return service;
    }

    /**
     * 创建0200数据清洗明细落盘服务
     */
    public SaveDataCleaning0200DetailService createSave0200DataCleaningDetailService() {
        String type = config.getProperty("save.0200.datacleaningdetail.bigdata.type");
        if (StringUtils.equals("hive", type)) {
            SaveDataCleaning0200DetailServiceImpl saveDataCleaning0200DetailService =
                    new SaveDataCleaning0200DetailServiceImpl();
            setSaveDataBigDataService(saveDataCleaning0200DetailService, "save.0200.datacleaningdetail");
            return saveDataCleaning0200DetailService;
        }
        return null;
    }

    /**
     * 创建0f37数据清洗服务
     */

    public DataCleaning0F37Service createDataCleaningService(){
        DataCleaning0F37ServiceImpl service = new DataCleaning0F37ServiceImpl();
        service.setSpeed_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0f37.threshold.speed."+runningTypeEnum.name())));
        service.setRotation_Max_Threshold(Integer.parseInt(
                config.getProperty("load.datacleaning.0f37.threshold.rotation."+runningTypeEnum.name())));
        service.setAcceleration_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0f37.threshold.acceleration."+runningTypeEnum.name())));
        service.setAvgSpeed_Weight_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0f37.threshold.avgSpeedWeight."+runningTypeEnum.name())));
        service.setMidSpeed_Weight_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.0f37.threshold.midSpeedWeight."+runningTypeEnum.name())));
        service.setLatLon_Max_Threshold(Double.parseDouble(
                config.getProperty("load.datacleaning.threshold.LatLonPerSec."+runningTypeEnum.name())));
        return service;
    }

    /**
     * 创建数据清洗明细落盘服务
     */
    public SaveDataCleaning0F37DetailService createSaveDataCleaningDetailService() {
        String type = config.getProperty("save.datacleaningdetail.bigdata.type");
        if (StringUtils.equals("hive", type)) {
            SaveDataCleaning0F37DetailServiceImpl saveDataCleaningDetailService = new SaveDataCleaning0F37DetailServiceImpl();
            setSaveDataBigDataService(saveDataCleaningDetailService, "save.datacleaningdetail");
            return saveDataCleaningDetailService;
        }
        return null;
    }

    /**
     * 创建静态路网数据匹配服务
     * @return
     */
    public RoadMapService createRoadMapService(){
        RoadMapServiceImpl service = new RoadMapServiceImpl();
        service.setMapUrl(config.getProperty("load.map.data.basepath"));
        service.setGpsType(Integer.parseInt(config.getProperty("load.map.data.gpsType")));
        return service;
    }
}
